package com.example.transformation;

import com.example.bean.WaterSensor;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * Created with IntelliJ IDEA.
 * ClassName: MapDemo
 * Package: com.example.transformation
 * Description:
 * User: fzykd
 *
 * @Author: LQH
 * Date: 2023-07-18
 * Time: 16:03
 */

public class MapDemo {
    public static void main(String[] args) throws Exception {

        //转换算子 map 映射
        //需求 提取WaterSensor中的id字段的功能

        //1.创建环境
        StreamExecutionEnvironment env =
                StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        //fromElements 是直接赋值数据
        DataStreamSource<WaterSensor> data = env.fromElements(
                //赋值WaterSensor对象的数据
                new WaterSensor("s1", 1L, 1),
                new WaterSensor("s2", 2L, 2),
                new WaterSensor("s3", 3L, 3)
        );

        //获取数据之后 获取数据ID
        //直接调用map 特点就是一进一出
        //参数要传一个MapFunction
        //这个匿名内部类的写法
//        SingleOutputStreamOperator<String> map = data.map(new MapFunction<WaterSensor, String>() {
//            @Override
//            public String map(WaterSensor waterSensor) throws Exception {
//                return waterSensor.getId();
//            }
//        });

        //lambda表达式
        //SingleOutputStreamOperator<String> map = data.map(val -> val.getId());


        //创建一个类 实现接口 匿名的话 可能其他类也会用到这个数据转换的逻辑

        SingleOutputStreamOperator<Object> map = data.map(new MyMap());
        map.print();
        env.execute();

    }

    public static class MyMap implements MapFunction<WaterSensor,Object> {
        @Override
        public String map(WaterSensor waterSensor) throws Exception {
            return waterSensor.getId();
        }
    }
}
